package com.example.mqtt.handler;
import com.example.mqtt.config.Topics;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

@Slf4j
public class EMQMessageHandler implements MessageHandler {


    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    KafkaSendResultHandler kafkaSendResultHandler;

    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
        if (message != null && message.getHeaders() != null) {
            String topicStr = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC).toString();
            log.info("handleMessage --->>> topic:{}",topicStr);
            if (topicStr.contains("gpd/p/04/")) {
                kafkaTemplate.send(Topics.DEVICE_DATA_PROTO_BUF4, topicStr, message.getPayload());
            } else {

            }
        } else {
            log.error("收到 unknow 消息");
        }
    }
}
